package com.ydl.learning.flink.demo

import com.ydl.learning.flink.demo.myFunction.PrintFunction
import com.ydl.learning.flink.demo.mySink.LogSink
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * stream demo
 *
 * @author ydl
 * @since 2020/6/24
 */
object StreamDemo extends Utils {

  case class WordWithCount(word: String, count: Long)

  def main(args: Array[String]): Unit = {
    val info = getSocketTextStream(9999)
    //设置全局并行度
    info._1.setParallelism(2)
    //获取数据: 从socket中获取
    val textDataStream = info._2
    val tupDataStream = textDataStream
      .map(new PrintFunction).name("print")
      .flatMap(_.split(" "))
      .map(WordWithCount(_, 1))

    //groupby: 按照指定的字段聚合
    val windowDstram = tupDataStream
      .keyBy("word")
      //窗口bsize=5秒, slid=1s
      //.timeWindow(Time.seconds(5), Time.seconds(5))
      .timeWindow(Time.seconds(5))

    windowDstram.sum("count").addSink(new LogSink[WordWithCount]()).name("logSink")

    //启动执行器，执行程序
    info._1.execute("Socket Window WordCount")
  }
}